package pers.cfeng.server.indexManager;


import lombok.Builder;
import pers.cfeng.common.DatabaseException;
import pers.cfeng.common.SubArray;
import pers.cfeng.server.dataManage.dataItem.DataItem;
import pers.cfeng.server.txManage.TransactionConstant;
import pers.cfeng.server.utils.ByteBufferParser;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * IM提供基于B+树的聚簇索引， 支持基于索引查找数据， 索引直接插入到数据库文件中
 * 不依赖VM
 * 索引和数据一样都是存储在DataItem中的， 一个二叉树的Node是存储在一个dataItem中的
 * [LeafFlag][KeyNumber][SiblingUid]   leafFlag标记是否为叶子， keyNumber为该节点中key的个数，siblingUid是兄弟节点存储在DM中的UID，
 * 后续就是存储的子节点和KeyN， 最后的KeyN始终都是MAX_VALUE，方便查找
 * [son0][key0][son1][key1]....[sonN][keyN]
 *
 * Node持有其B+树结构的引用， 还有底层的DataItem，基于Node可以快速操作数据
 *
 * leafFlag 1位  keyNumbers 2位  siblingUid 8位   son 8位 key 8 位
 */
@Builder
public class Node {
    static final int OF_IS_LEAF = 0;
    static final int OF_KEY_NUMBERS = OF_IS_LEAF + 1;
    static final int OF_SIBLING_UID = OF_KEY_NUMBERS + 2;
    static final int NODE_HEADER_SIZE = OF_SIBLING_UID + 8; //整个NOde的头长度

    static final int BALANCE_NUMBER = 32;
    //整个长度为头 + 32 * 2 + 2对[son][key]
    static final int NODE_SIZE = NODE_HEADER_SIZE + (2 * 8) * (BALANCE_NUMBER * 2 + 2);

    private BPlusTree tree;
    private DataItem dataItem;
    private SubArray raw;
    private long uid;

    //next结果封装内部类
    class SearchNextRes {
        long uid;
        long siblingUid;
    }

    class LeafSearchRangeRes {
        List<Long> uids;
        long siblingUid;
    }

    class InsertAndSplitRes {
        long siblingUid;
        long newSon;
        long newKey;
    }

    class SplitRes {
        long newSon;
        long newKey;
    }

    static void setRawIsLeaf(SubArray raw, boolean isLeaf) {
        if(isLeaf) {
            raw.raw[raw.start + OF_IS_LEAF] = (byte)1;
        } else
            raw.raw[raw.start + OF_IS_LEAF] = (byte) 0;
    }

    static boolean getRawIsLeaf(SubArray raw) {
        return raw.raw[raw.start + OF_IS_LEAF] == (byte) 1;
    }

    static void setRawNumKeys(SubArray raw, int numKeys) {
        System.arraycopy(ByteBufferParser.short2Byte((short) numKeys),0, raw.raw, raw.start + OF_KEY_NUMBERS, 2);
    }

    static int getRawNumKeys(SubArray raw) {
        return ByteBufferParser.parseShort(Arrays.copyOfRange(raw.raw, raw.start + OF_KEY_NUMBERS,raw.start + OF_SIBLING_UID));
    }

    static void setRawSibling(SubArray raw, long sibling) {
        System.arraycopy(ByteBufferParser.long2Byte(sibling),0, raw.raw, raw.start + OF_SIBLING_UID, 8);
    }

    static long getRawSibling(SubArray raw) {
        return ByteBufferParser.parseLong(Arrays.copyOfRange(raw.raw, raw.start + OF_SIBLING_UID,raw.start + OF_SIBLING_UID + 8));
    }

    //kth对key和son, 将uid插入到kth对之后，作为kth + 1对的son
    static void setRawKthSon(SubArray raw, long uid, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth * (8 * 2);
        System.arraycopy(ByteBufferParser.long2Byte(uid),0, raw.raw,offset, 8);
    }

    static long getRawKthSon(SubArray raw, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth* (8 * 2);
        return ByteBufferParser.parseLong(Arrays.copyOfRange(raw.raw, offset, offset + 8));
    }

    //kth对key和son, 将uid插入到kth对之后，作为kth + 1对的key
    static void setRawKthKey(SubArray raw, long uid, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth * (8 * 2) + 8;
        System.arraycopy(ByteBufferParser.long2Byte(uid),0, raw.raw,offset, 8);
    }

    static long getRawKthKey(SubArray raw, int kth) {
        int offset = raw.start + NODE_HEADER_SIZE + kth* (8 * 2);
        return ByteBufferParser.parseLong(Arrays.copyOfRange(raw.raw, offset, offset + 8));
    }

    //将kth对son key复制到to中
    static void copyRawFromKth(SubArray from, SubArray to, int kth) {
        int offset = from.start + NODE_HEADER_SIZE + kth* (8 * 2);
        System.arraycopy(from.raw, offset, to.raw, to.start + NODE_HEADER_SIZE, from.end - offset);
    }

    static void shiftRawKth(SubArray raw, int kth) {
        int begin = raw.start + NODE_HEADER_SIZE + (kth + 1) * (8 * 2);
        int end = raw.start + NODE_SIZE - 1;
        for(int i = end; i >= begin; i --) {
            raw.raw[i] = raw.raw[i - (8 * 2)];
        }
    }

    //新建root
    static byte[] newRootRaw(long left, long right, long key) {
        SubArray raw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);

        setRawIsLeaf(raw, false);
        setRawNumKeys(raw, 2);
        setRawSibling(raw, 0);
        setRawKthSon(raw, left, 0);
        setRawKthKey(raw, key, 0);
        setRawKthSon(raw, right, 1);
        setRawKthKey(raw, Long.MAX_VALUE, 1);

        return raw.raw;
    }

    static byte[] newNilRootRaw() {
        SubArray raw = new SubArray(new byte[NODE_SIZE],0, NODE_SIZE);

        setRawIsLeaf(raw, true);
        setRawNumKeys(raw, 0);
        setRawSibling(raw, 0);

        return raw.raw;
    }

        //装载load
    static Node loadNode(BPlusTree bPlusTree, long uid) throws Exception {
        DataItem di = bPlusTree.getDataManager().read(uid);
        if(di == null)
            throw new DatabaseException("node di is null");
        return Node.builder()
                .tree(bPlusTree)
                .dataItem(di)
                .raw(di.data())
                .uid(uid)
                .build();
    }

    public void release() {
        dataItem.release();
    }

    public boolean isLeaf() {
        dataItem.rLock();
        try {
            return getRawIsLeaf(raw);
        } finally {
            dataItem.rUnLock();
        }
    }

    public SearchNextRes searchNext(long key) {
        dataItem.rLock();
        try {
            SearchNextRes res = new SearchNextRes();
            int noKeys = getRawNumKeys(raw);
            for(int i = 0; i < noKeys; i ++) {
                long ik = getRawKthKey(raw, i);
                if(key < ik) {
                    res.uid = getRawKthSon(raw, i);
                    res.siblingUid = 0;
                    return res;
                }
            }
            res.uid = 0;
            res.siblingUid = getRawSibling(raw);
            return res;

        } finally {
            dataItem.rUnLock();
        }
    }

    public LeafSearchRangeRes leafSearchRange(long leftKey, long rightKey) {
        dataItem.rLock();
        try {
            int numKeys = getRawNumKeys(raw);
            int kth = 0;
            while (kth < numKeys) {
                long indexK = getRawKthKey(raw, kth);
                if(indexK >= leftKey)
                    break;
                kth ++;
            }
            List<Long> uids = new ArrayList<>();
            while (kth < numKeys) {
                long ik = getRawKthKey(raw, kth);
                if(ik <= rightKey) {
                    uids.add(getRawKthSon(raw, kth));
                    kth ++;
                } else {
                    break;
                }
            }
            long siblingUid = 0;
            if(kth == numKeys) {
                siblingUid = getRawSibling(raw);
            }
            LeafSearchRangeRes res = new LeafSearchRangeRes();
            res.uids = uids;
            res.siblingUid = siblingUid;
            return res;
        } finally {
            dataItem.rUnLock();
        }
    }

    public InsertAndSplitRes insertAndSplit(long uid, long key) throws Exception {
        boolean success = false;
        Exception err = null;
        InsertAndSplitRes res = new InsertAndSplitRes();

        dataItem.before();
        try {
            success = insert(uid, key);
            if(!success) {
                res.siblingUid = getRawSibling(raw);
                return res;
            }
            if(needSplit()) {
                try {
                    SplitRes r = split();
                    res.newSon = r.newSon;
                    res.newKey = r.newKey;
                    return res;
                } catch(Exception e) {
                    err = e;
                    throw e;
                }
            } else {
                return res;
            }
        } finally {
            if(err == null && success) {
                dataItem.after(TransactionConstant.NONE_TID);
            } else {
                dataItem.unBefore();
            }
        }
    }

    private boolean insert(long uid, long key) {
        int noKeys = getRawNumKeys(raw);
        int kth = 0;
        while(kth < noKeys) {
            long ik = getRawKthKey(raw, kth);
            if(ik < key) {
                kth ++;
            } else {
                break;
            }
        }
        if(kth == noKeys && getRawSibling(raw) != 0) return false;

        if(getRawIsLeaf(raw)) {
            shiftRawKth(raw, kth);
            setRawKthKey(raw, key, kth);
            setRawKthSon(raw, uid, kth);
            setRawNumKeys(raw, noKeys+1);
        } else {
            long kk = getRawKthKey(raw, kth);
            setRawKthKey(raw, key, kth);
            shiftRawKth(raw, kth+1);
            setRawKthKey(raw, kk, kth+1);
            setRawKthSon(raw, uid, kth+1);
            setRawNumKeys(raw, noKeys+1);
        }
        return true;
    }

    private boolean needSplit() {
        return BALANCE_NUMBER * 2 == getRawNumKeys(raw);
    }

    private SplitRes split() throws Exception {
        SubArray nodeRaw = new SubArray(new byte[NODE_SIZE], 0, NODE_SIZE);
        setRawIsLeaf(nodeRaw, getRawIsLeaf(raw));
        setRawNumKeys(nodeRaw, BALANCE_NUMBER);
        setRawSibling(nodeRaw, getRawSibling(raw));
        copyRawFromKth(raw, nodeRaw, BALANCE_NUMBER);
        long son = tree.getDataManager().insert(TransactionConstant.NONE_TID, nodeRaw.raw);
        setRawNumKeys(raw, BALANCE_NUMBER);
        setRawSibling(raw, son);

        SplitRes res = new SplitRes();
        res.newSon = son;
        res.newKey = getRawKthKey(nodeRaw, 0);
        return res;
    }

    //节点打印重写String
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("Is leaf: ").append(getRawIsLeaf(raw)).append("\n");
        int KeyNumber = getRawNumKeys(raw);
        sb.append("KeyNumber: ").append(KeyNumber).append("\n");
        sb.append("sibling: ").append(getRawSibling(raw)).append("\n");
        for(int i = 0; i < KeyNumber; i ++) {
            sb.append("son: ").append(getRawKthSon(raw, i)).append(", key: ").append(getRawKthKey(raw, i)).append("\n");
        }
        return sb.toString();
    }
}